package com.it.operator;

import com.it.operator.utils.SourceUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Operator_FlatMap {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(2);
        DataStreamSource<String> eventSource = SourceUtils.getStringSource(executionEnvironment);
        SingleOutputStreamOperator<String> flatMapedSource = eventSource.flatMap((FlatMapFunction<String, String>) (value, out) -> {
            String[] data = value.split(" ");
            for (String d : data) {
                out.collect(d);
            }
        }).returns(new TypeHint<String>() {
        });
        flatMapedSource.print();
        executionEnvironment.execute();

    }
}
